package pubsub

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"sync"
	"time"

	"device-admin/config/redis"

	"device-admin/internal/define"

	gredis "github.com/go-redis/redis/v8"

	"192.168.1.75/go-pkg/errorx"
	"192.168.1.75/go-pkg/logx"
)

var (
	RobotCMDChanMap sync.Map // map[string]chan string 设备命令channel
	pubSub          *gredis.PubSub
	subInitFinished = false // 订阅初始化标记
)

type DeviceCMDPayloadType struct {
	RobotID      string `json:"robot_id"`
	TerminalType string `json:"terminal_type"`
	CMDType      string `json:"cmd_type"`
	Payload      string `json:"payload"`
}

// CMDPublish 通过redis发布消息
// topic 主题
// payload 负载
func CMDPublish(robotID, terminalType, cmdType, payloadHex string) error {
	if robotID == "" {
		err := errorx.New("设备id不能为空", -1)
		logx.Error(context.Background(), "CMDPublish", logx.Err(err))
		return err
	}
	topic := define.RedisDeviceCmd + robotID + ":" + terminalType
	rdb := redis.NewDB()
	redisCtx, redisCtxCancel := context.WithTimeout(context.Background(), time.Second*3)
	defer redisCtxCancel()
	jsonString, _ := json.Marshal(DeviceCMDPayloadType{
		RobotID:      robotID,
		TerminalType: terminalType,
		CMDType:      cmdType,
		Payload:      payloadHex,
	})
	_, err := rdb.Publish(redisCtx, topic, string(jsonString)).Result()
	if err != nil {
		logx.Error(context.Background(), "CMDPublish", logx.Err(err))
		return nil
	}
	logx.Info(context.Background(), "CMDPublish", logx.String("robot_id", robotID), logx.String("terminal_type", terminalType), logx.String("cmdType", cmdType))
	return nil
}

// 发布命令响应结果
func PublishCMDRsp(robotID, terminalType, cmd, cmdSeq string, data []byte) error {
	topic := define.CMDRsp + robotID + ":" + terminalType + ":" + cmd + ":" + cmdSeq
	rdb := redis.NewDB()
	redisCtx, redisCtxCancel := context.WithTimeout(context.Background(), time.Second*3)
	defer redisCtxCancel()
	channels, _ := rdb.PubSubChannels(redisCtx, topic).Result()
	if len(channels) > 0 {
		payloadHex := base64.StdEncoding.EncodeToString(data)
		_, err := rdb.Publish(redisCtx, topic, payloadHex).Result()
		if err != nil {
			logx.Error(context.Background(), "publish", logx.Err(err))
			return err
		}
	}
	return nil
}

// CMD95SubHandleInit 订阅消息处理初始化
// 统一进行订阅，放入robotChan
func SubHandleInit() {
	if pubSub == nil {
		return
	}
	if subInitFinished {
		return
	}
	subInitFinished = true
	// mildmsg订阅
	go func() {
		defer func() {
			if err := recover(); err != nil {
				logx.Error(context.Background(), "SubMsgHandleRecover")
			}
		}()
		for msg := range pubSub.Channel() {
			var cmdPayload DeviceCMDPayloadType
			json.Unmarshal([]byte(msg.Payload), &cmdPayload)
			data := cmdPayload.Payload    // 实际的负载数据
			robotID := cmdPayload.RobotID // 机器人id
			cmdType := cmdPayload.CMDType // 命令类型
			terminalType := cmdPayload.TerminalType
			// 订阅到的信息，放入map，由各个设备会话消费
			chanMapKey := cmdType + "@" + robotID + "@" + terminalType
			logx.Info(context.Background(), "receive submsg", logx.String("chanMapKey", chanMapKey))
			go func() {
				RobotCMDChanMap.LoadOrStore(chanMapKey, make(chan string, 1))
				robotCMDChanInterface, _ := RobotCMDChanMap.Load(chanMapKey)
				robotCMDChan, _ := robotCMDChanInterface.(chan string)
				// 清理通道
				if len(robotCMDChan) > 0 {
					robotCMDChan = make(chan string, 1)
				}
				robotCMDChan <- data
			}()
		}
	}()
}

// AddSub 新增订阅
func AddSub(robotID, terminalType string) {
	// 主题
	topic := define.RedisDeviceCmd + robotID + ":" + terminalType
	// 订阅主题
	rdb := redis.NewDB()
	redisCtx, redisCtxCancel := context.WithTimeout(context.Background(), time.Second*3)
	defer redisCtxCancel()
	// 初始化pubsub
	if pubSub == nil {
		pubSub = rdb.Subscribe(redisCtx)
	}
	// 订阅消息，重试10次
	for i := 0; i < 10; i++ {
		err := pubSub.Subscribe(redisCtx, topic)
		if err == nil {
			break
		} else {
			logx.Error(context.Background(), "pubsub AddSub failed", logx.String("topic", topic), logx.Err(err))
		}
		time.Sleep(time.Millisecond * 100)
	}
}

// cancelSub 取消订阅
func CancelSub(robotID, terminalType string) {
	// 主题
	topic := define.RedisDeviceCmd + robotID + ":" + terminalType
	// 取消订阅
	redisCtx, redisCtxCancel := context.WithTimeout(context.Background(), time.Second*3)
	defer redisCtxCancel()
	if pubSub != nil {
		pubSub.Unsubscribe(redisCtx, topic)
	}
}
